package com.alison.tableapisql.chapter3_func;

import com.alison.tableapisql.chapter1_tableapiandsql.model.SensorReading;
import com.google.common.math.DoubleMath;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author : Ashiamd email: ashiamd@foxmail.com
 * @date : 2021/2/4 4:24 AM
 */
public class E5_UdfTest3_AggregateFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String filePath = "D:\\workspace\\lab\\learnbigdata\\learnflink\\flink-datastream\\src\\main\\resources\\tableapi\\E1.txt";

        // 1. 读取数据
        DataStream<String> inputStream = env.readTextFile(filePath);

        // 2. 转换成POJO
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<SensorReading>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(((element, recordTimestamp) -> element.timestamp * 1000)));

        // 3. 将流转换成表
        Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp");

        // 4. 自定义聚合函数，求当前传感器的平均温度值
        // 4.1 table API
        Top2 top2 = new Top2();

        sensorTable.printSchema();

        // 需要在环境中注册UDF
        tableEnv.registerFunction("top2", top2);
        Table resultTable = sensorTable
                .groupBy("id")
                .flatAggregate("top2(temp) as (tempv, rank)")
                .select("id, tempv, rank");

        // 4.2 SQL
        tableEnv.createTemporaryView("sensor", sensorTable);
        // 报错，无法识别 top2吧，  SQL parse failed. Encountered "(" at line 1, column 26.
//        Table resultSqlTable = tableEnv.sqlQuery("select id, top2(temp) as (tempv, rank) " +
//                " from sensor group by id ");
        Table resultSqlTable = tableEnv.sqlQuery("select id, temp, row_num" +
                " from ( " +
                "   select *, " +
                "     ROW_NUMBER() OVER (PARTITION BY id ORDER BY temp  DESC) as row_num " +
                "   from sensor )" +
                " WHERE row_num <= 2 ");

        // 打印输出
        tableEnv.toRetractStream(resultTable, Row.class).print("result");
        tableEnv.toRetractStream(resultSqlTable, Row.class).print("sql");
        // 改成 over(partition by XX order by

        env.execute();
    }


    public static class Top2Accum{
        public Double first;
        public Double second;
    }
    // 实现自定义的TableAggregateFunction, 求 第一高和第二高的温度, 输出 (hight, rank)
    public static class Top2 extends TableAggregateFunction<Tuple2<Double, Integer>, Tuple2<Double, Double>> {
        public void emitValue(Tuple2<Double, Double> acc, Collector<Tuple2<Double, Integer>> out) {
            out.collect(Tuple2.of(acc.f0, 1));
            out.collect(Tuple2.of(acc.f1, 2));
        }

        @Override
        public Tuple2<Double, Double> createAccumulator() {
            return new Tuple2<>(0.0, 0.0);
        }

        // 必须实现一个accumulate方法，来数据之后更新状态
        // 这里方法名必须是这个，且必须public。
        // 累加器参数，必须得是第一个参数；随后的才是我们自己传的入参
        public void accumulate(Tuple2<Double, Double> accumulator, Double temp) {
            if (temp > accumulator.f0) {
                accumulator.f1 = accumulator.f0;
                accumulator.f0 = temp;
            } else if (temp > accumulator.f1) {
                accumulator.f1 = temp;
            }
        }
    }

    /*
result> (true,+I[sensor_1, 35.8, 1])
result> (true,+I[sensor_1, 0.0, 2])
result> (true,+I[sensor_6, 15.4, 1])
result> (true,+I[sensor_6, 0.0, 2])
result> (true,+I[sensor_7, 6.7, 1])
result> (true,+I[sensor_7, 0.0, 2])
result> (true,+I[sensor_10, 38.1, 1])
result> (true,+I[sensor_10, 0.0, 2])
result> (false,-D[sensor_1, 35.8, 1])
result> (false,-D[sensor_1, 0.0, 2])
result> (true,+I[sensor_1, 36.3, 1])
result> (true,+I[sensor_1, 35.8, 2])
result> (false,-D[sensor_1, 36.3, 1])
result> (false,-D[sensor_1, 35.8, 2])
result> (true,+I[sensor_1, 36.3, 1])
result> (true,+I[sensor_1, 35.8, 2])
result> (false,-D[sensor_1, 36.3, 1])
result> (false,-D[sensor_1, 35.8, 2])
result> (true,+I[sensor_1, 37.1, 1])
result> (true,+I[sensor_1, 36.3, 2])

     */
}